package net.bwie.realtime.jtp.order.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

/**
 * 简介说明: 从 Kafka 读取订单明细数据，过滤 null 数据，统计各省份各窗口订单数和订单金额，关联省份维度和大区维度，最后将数据写入 Doris 交易域省份粒度下单各窗口汇总表。
 *
 * @author: LiLi
 * @date: 2025/06/03 21:04:42
 * @version: 1.0
 */
public class JtpOrderProvinceAggregateDwsJob {
    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv();

        // 2. 输入表-input：映射到Kafka消息队列
        createInputTable(tabEnv);

    }



    private static void createInputTable(TableEnvironment tabEnv) {
        TableResult tableResult = tabEnv.executeSql(
                "CREATE TABLE dwd_order_detail_kafka_source (\n" +
                        "    `id` STRING COMMENT 'ID',\n" +
                        "    `user_id` STRING COMMENT '用户ID',\n" +
                        "    `process_time` TIMESTAMP(3) COMMENT '数据处理时间',\n" +
                        "    `search_terms` STRING COMMENT '搜索关键词',\n" +
                        "    `brand_pref` STRING COMMENT '品牌偏好(奢侈品牌等)',\n" +
                        "    `price_sensitivity` DOUBLE COMMENT '价格敏感度(0-1)',\n" +
                        "    `device_info` STRING COMMENT '设备信息(Mac等)',\n" +
                        "    `constellation` STRING COMMENT '星座',\n" +
                        "    `category_pref` STRING COMMENT '品类偏好(汽车用品等)',\n" +
                        "    `age_label` STRING COMMENT '年龄标签(50岁以上等)',\n" +
                        "    `time_behavior` STRING COMMENT '时间行为数据(早晨、下午、夜间活动)',\n" +
                        "    `social_interaction` INT COMMENT '社交互动分数',\n" +
                        "    `gender` STRING COMMENT '性别(女性用户等)',\n" +
                        "    `height` DOUBLE COMMENT '身高(cm)',\n" +
                        "    `weight` DOUBLE COMMENT '体重(kg)',\n" +
                        "    `source` STRING COMMENT '数据来源(top_level, health_device等)',\n" +
                        "    `source_label` STRING COMMENT '来源标签',\n" +
                        "    `source_weight` DOUBLE COMMENT '来源权重',\n" +
                        "    `dynamic_fields` STRING COMMENT '动态扩展字段',\n" +
                        "    `create_time` TIMESTAMP(3) METADATA FROM 'timestamp' COMMENT '创建时间',\n" +
                        "    `proc_time` AS PROCTIME(),\n" +
                        "    WATERMARK FOR create_time AS create_time\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'user_profile_tags',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "    'properties.group.id' = 'gid-dws-trade-sku-order',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json',\n" +
                        "    'json.fail-on-missing-field' = 'false',\n" +
                        "    'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );

        // 然后可以执行查询并打印
        tabEnv.executeSql("SELECT * FROM dwd_order_detail_kafka_source").print();

    }


    private static TableEnvironment getTableEnv() {
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // 3返回对象
        return tabEnv;
    }
}
